package org.ala.tree.lsm.sstable;

import java.io.IOException;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.nio.channels.FileChannel.MapMode;
import java.nio.file.Path;
import java.nio.file.StandardOpenOption;
import java.util.List;
import java.util.Map.Entry;

import org.ala.tree.lsm.KV;
import org.ala.tree.lsm.LSMConfig;
import org.ala.tree.lsm.sstable.block.DataBlock;

/**
 * SSTable搜索工具
 *
 * @author ala
 * @date 2025年5月30日
 */
public class SSTableSearcher<V> {

	
	/**
	 * 	每一层的SSTable
	 */
	protected List<SSTable<V>>[] levelSSTables;
	/**
	 * 	
	 */
	protected LSMConfig config;
	/**
	 * 	读取工具
	 */
	protected SSTableReader<V> reader;
	
	
	@SuppressWarnings("unchecked")
	public SSTableSearcher(LSMConfig config) {
		this.config = config;
		levelSSTables = new List[config.countLevel];
		reader = new SSTableReader<>();
	}
	
	
	/**
	 * 	从level开始，搜索某个key
	 * 	<p>	如果找不到往上逐层搜索
	 * @throws IOException 
	 */
	public V search(String key, int level, Class<V> vclass) throws IOException {
		if (level >= config.countLevel) {return null;}
		if (levelSSTables[level] == null) {
			loadSSTable(level, vclass);
		}
		if (levelSSTables[level] == null) {return null;}
		V v = searchSSTable(key, level);
		if (v == null) {v = search(key, level+1, vclass);}
		return v;
	}
	/**
	 * 	加载第i层的SSTable
	 * 	@throws IOException 
	 */
	protected void loadSSTable(int level, Class<V> vclass) throws IOException {
		levelSSTables[level] = reader.read(config, level, vclass);
	}
	/**
	 * 	在第i层中，搜索key
	 * @throws IOException 
	 */
	protected V searchSSTable(String k, int level) throws IOException {
		for (int i = 0 ; i < levelSSTables[level].size() ; i++) {
			SSTable<V> sstable = levelSSTables[level].get(i);
			V v = searchSSTable(k, sstable);
			if (v != null) {return v;}
		}
		return null;
	}
	protected V searchSSTable(String k, SSTable<V> sstable) throws IOException {
		//	布隆过滤器如果不存在，那就真的不存在。如果存在，有可能是假阳
		if (sstable.getBloomFilter().exists(k)) {
			if (sstable.getIndex() == null) {
				reader.loadIndex(config, sstable);
			}
			Entry<String, Integer> e = sstable.getIndex().floorEntry(k);
			if (e != null) {
				if (sstable.getDataBlocks().get(e.getValue()) == null) {
					sstable.getDataBlocks().put(e.getValue(), loadDataBlock(sstable.getLevel(), sstable.getId(), e.getValue(), sstable.getDataBlockCapacity(), sstable.getVclass()));
				}
				DataBlock<V> db = sstable.getDataBlocks().get(e.getValue());
				db.resetIteratprPos();
				while (db.hasNext()) {
					KV<V> kv = db.next();
					if (kv.getKey().equals(k)) {return kv.getVal();}
				}
			}
		}
		return null;
	}
	protected DataBlock<V> loadDataBlock(int level, String id, int offset, int dataBlockCapacity, Class<V> vclass) throws IOException {
		FileChannel fileChannel = FileChannel.open(Path.of(String.format("%s/%s%d/%s.sstable", config.ROOT_DIR, config.levelPrefix, level, id)), StandardOpenOption.READ);
		//	先读头
		MappedByteBuffer headBuf = fileChannel.map(MapMode.READ_ONLY, offset, 4);
		int len = headBuf.getInt();
		//	把内容读出来
		MappedByteBuffer dataBuf = fileChannel.map(MapMode.READ_ONLY, offset+4, len);
		return DataBlock.decode(dataBlockCapacity, dataBuf, vclass);
	}
	
	
	
}
